package com.atguigu.flink.timer;

import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2022/11/25
 *
 *  去重: 去掉同一种传感器重复的水位值
 */
public class Demo8_MapState
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            .keyBy(WaterSensor::getId)
            .process(new KeyedProcessFunction<String, WaterSensor, String>()
            {

                private MapState<Integer, Object> vc;
                private ReducingState<WaterSensor> state;

                @Override
                public void open(Configuration parameters) throws Exception {
                    vc = getRuntimeContext().getMapState(new MapStateDescriptor<Integer, Object>("vc", Integer.class, Object.class));
                }

                @Override
                public void processElement(WaterSensor value,Context ctx, Collector<String> out) throws Exception {

                    //来一条数据，放入状态
                    vc.put(value.getVc(),"a");

                    //获取Map的key，都是不重复的水位
                    List<Integer> vcList = MyUtil.parseList(vc.keys());

                    out.collect(ctx.getCurrentKey() + ":" + vcList);

                }
            }).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
